package com.neusoft.databus.agent.interactive.collect;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.Reader;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.commons.net.ftp.FTPClient;

import com.google.common.collect.Maps;
import com.neusoft.bizcore.web.utils.SpringUtil;
import com.neusoft.databus.agent.parser.WangguanParser;
import com.neusoft.databus.agent.pool.wangguan.WangguanConnector;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.ResourceBean;
import com.neusoft.databus.common.connection.WangguanConnection;
import com.neusoft.databus.common.interactive.collect.Collector;
import com.neusoft.databus.common.parser.Parser;
import com.univocity.parsers.csv.CsvParser;
import com.univocity.parsers.csv.CsvParserSettings;

import lombok.extern.slf4j.Slf4j;

/**
 * description
 *
 * @author sunchf
 * @date 2019年4月2日 上午9:37:26
 */
@Slf4j
public class WangguanWangguanCollector extends Collector<FTPClient> {

    public WangguanWangguanCollector(final Parser parser) {
        super(parser);
        this.connector = SpringUtil.getBean(WangguanConnector.class);
    }

    @Override
    protected void doCollect(final CommandBean command) throws Exception {
        final CsvParser csvParser = this.csvParser();

        final Map<String, List<ResourceBean>> groups = Maps.newHashMap();
        command.getResourceBean().stream().filter(it -> it.getConfig().containsKey("type")).forEach(it -> {
            final String type = (String) it.getConfig().get("type");
            if (groups.containsKey(type)) {
                groups.get(type).add(it);
            } else {
                final List<ResourceBean> l = new ArrayList<>();
                l.add(it);
                groups.put(type, l);
            }
        });

        // 根据类型，分类所有的文件
        final Map<String, List<String>> typeFptFileNames = Maps.newHashMap();
        // 根据类型，存放每类最新的1个文件
        final Map<String, String> latestFiles = Maps.newHashMap();
        // 根据类型，解析最新文件里存放的所有数据行
        final Map<String, List<String[]>> typeRows = Maps.newHashMap();

        try {
            final WangguanConnection connection = (WangguanConnection) command.getConnection().asUsedConnection();
            this.conn = this.connector.getConn(connection);
            this.conn.changeWorkingDirectory(connection.getPath());

            final String[] ftpFullFileNames = this.conn.listNames(connection.getPath());
            final List<String> ftpFileNames = Stream.of(ftpFullFileNames)
                    .map(it -> it.replace(connection.getPath(), "")).collect(Collectors.toList());

            WangguanWangguanCollector.log.info("found {} at ftp server.",
                    ftpFileNames.stream().collect(Collectors.joining(", ", "[", "]")));
            // 根据类型，分类所有的文件
            ftpFileNames.forEach(n -> {
                final String type = n.split("_")[1];
                if (typeFptFileNames.containsKey(type)) {
                    typeFptFileNames.get(type).add(n);
                } else {
                    final List<String> l = new ArrayList<>();
                    l.add(n);
                    typeFptFileNames.put(type, l);
                }
            });

            if (WangguanWangguanCollector.log.isDebugEnabled()) {
                typeFptFileNames.keySet().forEach(k -> {
                    WangguanWangguanCollector.log.debug("{}: {}", k, typeFptFileNames.get(k).size());
                });
            }

            // 根据类型，存放每类最新的1个文件
            for (final Map.Entry<String, List<String>> entry : typeFptFileNames.entrySet()) {
                final List<String> names = entry.getValue();
                names.sort((a, b) -> b.compareTo(a));
                latestFiles.put(entry.getKey(), names.get(0));
            }
            if (WangguanWangguanCollector.log.isDebugEnabled()) {
                latestFiles.keySet().forEach(k -> {
                    WangguanWangguanCollector.log.debug("{}: {}", k, latestFiles.get(k));
                });
            }

            latestFiles.entrySet().stream().forEach(entry -> {
                final String fileName = entry.getValue();
                try {
                    final OutputStream outputStream =
                            new FileOutputStream(connection.getTmpPath() + "/" + fileName);

                    final boolean isSuccess = this.conn.retrieveFile(connection.getPath() + fileName, outputStream);
                    if (!isSuccess) {
                        WangguanWangguanCollector.log.warn("download file {} fail", fileName);
                    } else {
                        WangguanWangguanCollector.log.info("download file {} success to ", fileName,
                                connection.getTmpPath());
                    }

                    Reader reader = null;
                    try {
                        reader = new InputStreamReader(
                                new FileInputStream(connection.getTmpPath() + "/" + fileName), "UTF-8");
                    } catch (final UnsupportedEncodingException e) {
                        WangguanWangguanCollector.log
                                .warn("Unable to read file " + connection.getTmpPath() + "/" + fileName, e);
                    }
                    if (null == reader) {
                        return;
                    }

                    if (!typeRows.containsKey(entry.getKey())) {
                        typeRows.put(entry.getKey(), new ArrayList<>());
                    }
                    csvParser.beginParsing(reader);
                    String[] row;
                    while ((row = csvParser.parseNext()) != null) {
                        typeRows.get(entry.getKey()).add(row);
                    }
                    csvParser.stopParsing();

                } catch (final Exception e) {
                    WangguanWangguanCollector.log.error("", e);
                }
            });

            //            final FTPFile[] ftpFiles = this.conn.listFiles(connection.getPath());
            //
            //            WangguanWangguanCollector.log.info("there are {} files in ftp server.", ftpFiles.length);
            //
            //            final List<String> successParsedFiles = Lists.newArrayList();
            //
            //            for (final FTPFile ftpFile : ftpFiles) {
            //                final OutputStream outputStream =
            //                        new FileOutputStream(connection.getTmpPath() + "/" + ftpFile.getName());//创建文件输出流
            //
            //                final boolean isSuccess = this.conn.retrieveFile(ftpFile.getName(), outputStream);
            //                if (!isSuccess) {
            //                    WangguanWangguanCollector.log.warn("download file {} fail", ftpFile.getName());
            //                    continue;
            //                } else {
            //                    WangguanWangguanCollector.log.info("download file {} success to ", ftpFile.getName(),
            //                            connection.getTmpPath());
            //                }
            //
            //                Reader reader = null;
            //                try {
            //                    reader = new InputStreamReader(
            //                            new FileInputStream(connection.getTmpPath() + "/" + ftpFile.getName()), "UTF-8");
            //                } catch (final UnsupportedEncodingException e) {
            //                    WangguanWangguanCollector.log
            //                            .warn("Unable to read file " + connection.getTmpPath() + "/" + ftpFile.getName(), e);
            //                }
            //                if (null == reader) {
            //                    continue;
            //                }
            //
            //                final String type = ftpFile.getName().split("_")[1];
            //                if (!typeRows.containsKey(type)) {
            //                    typeRows.put(type, new ArrayList<>());
            //                }
            //                csvParser.beginParsing(reader);
            //                String[] row;
            //                while ((row = csvParser.parseNext()) != null) {
            //                    typeRows.get(type).add(row);
            //                }
            //                csvParser.stopParsing();
            //
            //                // 记录解析成功的文件
            //                successParsedFiles.add(ftpFile.getName());
            //            }
            //
            //            // 从FTP上删除解析成功的文件
            //            for (final String fileName : successParsedFiles) {
            //                final boolean delectedSuccess = this.conn.deleteFile(fileName);
            //                if (delectedSuccess) {
            //                    WangguanWangguanCollector.log.info("delete file {} sucess", fileName);
            //                } else {
            //                    WangguanWangguanCollector.log.warn("delete file {} fail", fileName);
            //                }
            //            }

            for (final Map.Entry<String, List<ResourceBean>> entry : groups.entrySet()) {
                final String type = entry.getKey();
                if (!typeRows.containsKey(type)) {
                    continue;
                }
                final List<String[]> csvRows = typeRows.get(type);

                if ("itportdata".equals(type)) {
                    entry.getValue().stream().forEach(resource -> {
                        final String slave = (String) resource.getConfig().get("slave");

                        final Optional<String[]> opt =
                                csvRows.stream().filter(row -> (row[0] + "@@" + row[2]).equals(slave)).findAny();
                        if (opt.isPresent()) {
                            resource.getMetrics().forEach(metric -> {
                                final String v = opt.get()[Integer.valueOf(metric.getCommand())];
                                metric.setOriginalValue(v);
                                metric.setValue(v);

                            });
                        }
                    });
                } else {
                    entry.getValue().stream().forEach(resource -> {
                        final String slave = (String) resource.getConfig().get("slave");
                        final Optional<String[]> opt =
                                csvRows.stream().filter(row -> row[0].equals(slave)).findAny();
                        if (opt.isPresent()) {
                            resource.getMetrics().forEach(metric -> {
                                final String v = opt.get()[Integer.valueOf(metric.getCommand())];
                                metric.setOriginalValue(v);
                                metric.setValue(v);

                            });
                        }
                    });
                }
            }

            try {
                ((WangguanParser) this.parser).parseCollectResponse(command);
            } catch (final Exception e) {
                WangguanWangguanCollector.log.error("parse wangguan metric error", e);
            }

        } catch (final Exception e) {
            throw e;
        } finally {
            try {
                this.conn.disconnect();
                this.conn.logout();
            } catch (final Exception e) {

            }
            try {
                if (WangguanWangguanCollector.log.isDebugEnabled()) {
                    WangguanWangguanCollector.log.debug("return connection object to poll, {}",
                            command.getConnection());
                }
                this.connector.returnConn(command.getConnection().asUsedConnection(), this.conn);
            } catch (final Exception e) {

            }
        }

    }

    private CsvParser csvParser() {
        final CsvParserSettings settings = new CsvParserSettings();
        settings.getFormat().setDelimiter("^");
        final CsvParser parser = new CsvParser(settings);
        return parser;
    }

}
